package com.behavioranalysis.flinkprogram.flink.ad;

import com.behavioranalysis.flinkprogram.dao.IAdBlacklistDAO;
import com.behavioranalysis.flinkprogram.dao.factory.DAOFactory;
import com.behavioranalysis.flinkprogram.domain.AdBlacklist;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午8:26
 * updated by yang on 20-2-14 下午8:26
 */
public class FilterBlackUserProcess extends
        ProcessAllWindowFunction<Tuple6<Long, String, String, String, String, String>, Tuple6<Long, String, String, String, String, String>, TimeWindow> {
    private static final long serialVersionUID = 1L;

    @Override
    public void process(Context context, Iterable<Tuple6<Long, String, String, String, String, String>> elements, Collector<Tuple6<Long, String, String, String, String, String>> out) throws Exception {
        // 将blacklist全部查出来
        IAdBlacklistDAO adBlacklistDAO = DAOFactory.getAdBlackListDAO();
        List<AdBlacklist> adBlacklists = adBlacklistDAO.findAll();

        List<String> blacklist = new ArrayList<>();
        for (AdBlacklist adBlacklist : adBlacklists) {
            blacklist.add(String.valueOf(adBlacklist.getUserid()));
        }

        //System.out.println("黑名单总计:" + blacklist.size());

        // 判断此窗口中的记录的userid是否在blacklist中，不在就output
        for (Tuple6<Long, String, String, String, String, String> tuple6 : elements) {
            String userid = tuple6.f3;
            if (!blacklist.contains(userid)) {
                // 2. 将此记录collect
                out.collect(tuple6);
            }
        }
    }
}
